package com.linshidream.intrachat.simplenio.packet.boots;

import com.linshidream.intrachat.simplenio.packet.context.ConnectionContext;
import com.linshidream.intrachat.simplenio.packet.handler.PacketReader;

import java.io.IOException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingDeque;

/**
 * Created on 2025/8/17 22:00
 *
 * @author linshidream
 * @version 1.0.0
 * @description
 */
public class PacketWorker implements Runnable {

    private final BlockingQueue<ConnectionContext> ctxs = new LinkedBlockingDeque<>();

    /**
     * 一个 PacketWorker 对象持有一个监听器
     */
    private final Selector selector;

    public Selector nextSelector(){
        return selector;
    }


    /**
     * 每个socket对象持有一个上下文连接
     */
    private final Map<SocketChannel, ConnectionContext> contextMap;


    public PacketWorker() throws IOException {
        this.selector = Selector.open();
        // 私有 Map，无需线程安全
        this.contextMap = new HashMap<>();
    }

    public void register(ConnectionContext ctx) {
        ctxs.offer(ctx);
        // 这里一定要唤醒 run里面就开始取 SocketChannel 去注册
        selector.wakeup();
    }

    @Override
    public void run() {
        try {
            while (true) {
                selector.select();

                // 会把所有的 SocketChannel 注册到 selector，直到注册完成
                ConnectionContext ctx;
                while ((ctx = ctxs.poll()) != null) {
                    SocketChannel sc = ctx.channel();
                    sc.configureBlocking(false);
                    sc.register(selector, SelectionKey.OP_READ);

                    // 建立连接
                    contextMap.put(sc, ctx);
                }

                Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
                while (iterator.hasNext()) {
                    SelectionKey selectionKey = iterator.next();
                    iterator.remove();

                    if (selectionKey.isValid() && selectionKey.isReadable()) {
                        SocketChannel client = (SocketChannel) selectionKey.channel();
                        ConnectionContext connectionContext = contextMap.get(client);
                        // 把通道交给解码器去解码成消息 支持累积缓冲区
                        PacketReader.doRead(connectionContext);
                    }

                    if (selectionKey.isValid() && selectionKey.isWritable()) {
                        SocketChannel client = (SocketChannel) selectionKey.channel();
                        ConnectionContext connectionContext = contextMap.get(client);
                        // 写监听 上次未写完的数据可以继续写（socket 缓冲区满了）
                        connectionContext.writer().doWrite();
                    }
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
            System.out.println("PacketWorker 线程执行异常," + Thread.currentThread().getName());
        }
    }

}
